package cn.ac.iie.di.ban.ycl.cleaner.handler;

import cn.ac.iie.di.ban.ycl.cleaner.CleanerMain;
import cn.ac.iie.di.ban.ycl.cleaner.common.Constants;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import org.apache.commons.configuration2.Configuration;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author zmc <zhangmingcheng@iie.ac.cn>
 * @date 2016-9-8 13:53:12
 * @version V2.0
 * @desc 数据去重
 */
public class CleanerHandlerForGroup implements Runnable {

    final Logger LOG = LoggerFactory.getLogger(CleanerHandlerForGroup.class);
    BlockingQueue<REMessage> queue;
    FilterInterface deduplicationFilter;
    ContentFilter contentFilter;
    CtTimeDataCountEveryDay cTTimeDataCountEveryDay;
    boolean isFilterNecessary;
    boolean isDeduplicationNecessary;

    public CleanerHandlerForGroup(Configuration configuration, BlockingQueue queue) {
        this.queue = queue;
        deduplicationFilter = new DeduplicationFilterForGroup(configuration);
        contentFilter = new ContentFilter(configuration);
        cTTimeDataCountEveryDay = new CtTimeDataCountEveryDay();
        isFilterNecessary = configuration.getBoolean(Constants.FILTER_NECESSARY);
        isDeduplicationNecessary = configuration.getBoolean(Constants.DEDUPLICATION_NECESSARY);
    }

    @Override
    public void run() {
        while (true) {
            try {
                REMessage message = queue.poll(3, TimeUnit.SECONDS);
                if (message == null) {
                    LOG.warn("there is not message from the receiver queue!");
                    continue;
                }
                LOG.info("receive messages num: {}", message.data.size());
                handle(message);
            } catch (Exception ex) {
                LOG.error("clean data has error! {}", ExceptionUtils.getFullStackTrace(ex));
            }
        }
    }

    private void handle(REMessage message) throws Exception {
        //根据ct统计数据量
        Map<String, Integer> cTTimeDataCount = cTTimeDataCountEveryDay.getcTTimeDataCount(message);
        CleanerMain.cTReceiveCountQueue.put(cTTimeDataCount);
        //内容过滤
        if (isFilterNecessary) {
            message = contentFilter.filter(message);
        }
        CleanerMain.filterQueue.put(message);
        //垃圾数据
        REMessage rubishMessage = contentFilter.getRubbishMessage();
        CleanerMain.rubbishQueue.put(rubishMessage);
        //数据去重
        if (isDeduplicationNecessary) {
            message = deduplicationFilter.filter(message);
        }
        CleanerMain.deduplicationQueue.put(message);
        //根据ct统计数据量
        cTTimeDataCount = cTTimeDataCountEveryDay.getcTTimeDataCount(message);
        CleanerMain.cTSendCountQueue.put(cTTimeDataCount);
    }
}
